import com.atguigu.util.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author hfh
 * @version 1.0
 * @date 2022/9/3 10:14
 */
public class test01 {
    public static void main(String[] args) throws Exception {
        //获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //设置状态后端
        /*env.enableCheckpointing(5*60*1000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setCheckpointTimeout(3*60*1000L);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        //获取topic_log数据
 /*       {
            "common": {
            "ar": "13",
                    "ba": "iPhone",
                    "ch": "Appstore",
                    "is_new": "1",
                    "md": "iPhone Xs Max",
                    "mid": "mid_1",
                    "os": "iOS 13.2.3",
                    "sc": "2",
                    "sid": "f5b26601-d182-484e-9ff5-6a084caa0c54",
                    "uid": "464",
                    "vc": "v2.1.134"
        },
            "displays": [
            {
                "display_type": "promotion",
                    "item": "7",
                    "item_type": "course_id",
                    "order": 1,
                    "pos_id": 5
            },
            {
                "display_type": "promotion",
                    "item": "5",
                    "item_type": "course_id",
                    "order": 2,
                    "pos_id": 1
            },
            {
                "display_type": "query",
                    "item": "1",
                    "item_type": "course_id",
                    "order": 3,
                    "pos_id": 2
            },
            {
                "display_type": "promotion",
                    "item": "2",
                    "item_type": "course_id",
                    "order": 4,
                    "pos_id": 3
            }
    ],
            "page": {
            "during_time": 8410,
                    "item": "17601",
                    "item_type": "order_id",
                    "last_page_id": "order",
                    "page_id": "payment"
        },
            "ts": 1662127931625
        }*/
        String topicLog = "topic_log";
        String groupId="teat01";
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.getKafkaConsumer(topicLog, groupId));

        streamSource.print();

        //获取topic_db数据
       /* {
            "database": "edu",
                "table": "test_exam_question",
                "type": "insert",
                "ts": 1662171752,
                "xid": 108819,
                "xoffset": 2,
                "data": {
            "id": 30750,
                    "exam_id": 2091,
                    "paper_id": 199,
                    "question_id": 1108,
                    "user_id": 21,
                    "answer": "3111",
                    "is_correct": "1",
                    "score": 5.60,
                    "create_time": "2022-09-03 10:22:33",
                    "update_time": null,
                    "deleted": "0"
        }
        }*/

        env.execute(groupId);
    }
}
